package cn.monkey.io.transport.netty;

import cn.monkey.io.transport.core.Exchange;
import cn.monkey.io.transport.core.ExchangeContext;
import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.SocketAddress;

public class DefaultChannelGroupExchange implements ChannelGroupExchange {

    private final String id;
    private final ChannelGroup channelGroup;
    private final ExchangeContext exchangeContext;
    private final ChannelGroupInbound inbound;
    private final ChannelGroupOutbound outbound;

    public DefaultChannelGroupExchange(String id, ChannelGroup channelGroup) {
        this.id = id;
        this.channelGroup = channelGroup;
        this.inbound = new DefaultChannelGroupInbound(channelGroup);
        this.outbound = new DefaultChannelGroupOutbound(channelGroup);
        this.exchangeContext = new DefaultChannelGroupExchangeContext(channelGroup);
    }


    @Override
    public String id() {
        return this.id;
    }

    @Override
    public ChannelGroup channelGroup() {
        return this.channelGroup;
    }

    @Override
    public ExchangeContext context() {
        return this.exchangeContext;
    }

    @Override
    public ChannelGroupInbound inbound() {
        return this.inbound;
    }

    @Override
    public ChannelGroupOutbound outbound() {
        return this.outbound;
    }

    @Override
    public Mono<Void> close() {
        Mono<Void> m1 = this.inbound.close();
        Mono<Void> m2 = this.outbound.close();
        return Flux.merge(m1, m2).then();
    }

    @Override
    public boolean isActive() {
        return channelGroup.stream().anyMatch(Channel::isActive);
    }

    @Override
    public ChannelGroupExchange bind() {
        this.exchangeContext.setAttribute(Exchange.KEY, this);
        return this;
    }

    @Override
    public Mono<Void> destroy(ChannelExchange channelExchange) {
        Mono<Void> destroy = channelExchange.destroy();
        return destroy.doFinally(signalType -> this.channelGroup.remove(channelExchange.channel()));
    }

    @Override
    public SocketAddress localAddress() {
        throw new UnsupportedOperationException();
    }

    @Override
    public SocketAddress remoteAddress() {
        throw new UnsupportedOperationException();
    }
}
